package com.atguigu.gmall.realtime.dwd.db.split.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TableProcessDwd;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSourceUtil;
import com.atguigu.gmall.realtime.dwd.db.split.function.BaseDbTableProcessFunction;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;


/**
 * ClassName: DwdBaseDb
 * Description:
 *
 * @Create 2024/4/22 15:07
 */
public class DwdBaseDb extends BaseApp {

    public static void main(String[] args) {
        new DwdBaseDb().start(10019, 4, "dwd_base_db", Constant.TOPIC_DB);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaSourceDS) {
        //TODO 对流中数据类型进行转换并ETL   Json -> JsonObject
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaSourceDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(jsonStr);
                            String type = jsonObject.getString("type");
                            /*if (!type.startsWith("bootstrap-")) {
                            }*/
                                collector.collect(jsonObject);
                        } catch (Exception e) {
                            throw new RuntimeException("不是一个标准json");
                        }
                    }
                }
        );

        //jsonObjDS.print("~~~");

        //TODO CDC读取配置表信息
        //创建MySqlSource对象
        MySqlSource<String> mySqlSource = FlinkSourceUtil.getMySqlSource("gmall1030_config", "gmall1030_config.table_process_dwd");

        //读取数据，封装为流
        DataStreamSource<String> mySqlStrDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql_source");

        //类型转换  JsonStr -> 实体类对象
        SingleOutputStreamOperator<TableProcessDwd> tpDS = mySqlStrDS.map(
                new MapFunction<String, TableProcessDwd>() {
                    @Override
                    public TableProcessDwd map(String jsonStr) throws Exception {

                        TableProcessDwd tableProcessDwd = null;

                        JSONObject jsonObj = JSON.parseObject(jsonStr);

                        String op = jsonObj.getString("op");

                        if ("d".equals(op)) {
                            //fastJson底层自动做了下划线 -> 驼峰命名转换
                            tableProcessDwd = jsonObj.getObject("before", TableProcessDwd.class);
                        } else {
                            tableProcessDwd = jsonObj.getObject("after", TableProcessDwd.class);
                        }
                        tableProcessDwd.setOp(op);

                        return tableProcessDwd;
                    }
                }
        );

        //TODO  广播配置表
        MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor = new MapStateDescriptor<>("mapStateDescriptor", String.class, TableProcessDwd.class);
        BroadcastStream<TableProcessDwd> broadcastDS = tpDS.broadcast(mapStateDescriptor);

        //TODO 将主流业务数据和广播数据关联
        BroadcastConnectedStream<JSONObject, TableProcessDwd> connectDS = jsonObjDS.connect(broadcastDS);

        //TODO 对关联的数据进行处理
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> realDS = connectDS.process(
                //匿名内部类方式可能会产生序列化问题，将方法封装提出
                new BaseDbTableProcessFunction(mapStateDescriptor)
        );

        //TODO 分流写入Kafka
        realDS.print();
        realDS.sinkTo(FlinkSinkUtil.getKafkaSink());

    }


}
